"""
Case Type   : SQL_DDL_Drop_Owned
Case Name   : 使用CASCADE级联删除所有依赖于被删除对象的对象
Create At   : 2022/08/24
Owner       : @li-xin12345
Description :
    1.创建用户
    2.用户1创建表1和视图1
    3.用户2创建依赖于表1的视图2
    4.使用CASCADE级联删除所有依赖于被删除对象的对象
    5.删除用户
Expect      :
    1.创建成功
    2.创建成功
    3.创建成功
    4.用户1所拥有的对象删除成功，且递归删除由用户2所拥有的对象
    5.删除成功
History     :
    Modified by @kyeleze 2022/11/17:优化用例，保证\d查询结果匹配
    Modified by @nanyang12 2024/12/31:研发代码变更，有视图依赖的对象可以被删除，修改用例预期以适配
"""
import os
import unittest
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro


class DropOwned(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.user_node = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.user_name_1 = 'u_drop_owned_0004_1'
        self.user_name_2 = 'u_drop_owned_0004_2'
        self.table_name_1 = 't_drop_owned_0004_1'
        self.view_name_1 = 'v_drop_owned_0004_1'
        self.view_name_2 = 'v_drop_owned_0004_2'

    def test_drop_owned(self):
        text = "-----step1:创建用户;expect:创建成功-----"
        self.logger.info(text)
        create_user = f"drop user if exists {self.user_name_1}," \
            f"{self.user_name_2};" \
            f"create user {self.user_name_1} " \
            f"sysadmin password '{macro.COMMON_PASSWD}';" \
            f"create user {self.user_name_2} " \
            f"sysadmin password '{macro.COMMON_PASSWD}';"
        create_msg = self.primary_sh.execut_db_sql(create_user)
        self.logger.info(create_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        self.assertEquals(2, create_msg.count(
            self.constant.CREATE_ROLE_SUCCESS_MSG), '执行失败' + text)

        text = "-----step2:用户1创建表1和视图1;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop table if exists {self.table_name_1};" \
            f"create table public.{self.table_name_1}" \
            f"(id int, name varchar(15));" \
            f"insert into {self.table_name_1} values(1, '软件开发');" \
            f"drop view if exists {self.view_name_1};" \
            f"create view {self.view_name_1} as " \
            f"select * from {self.table_name_1};"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name_1} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertNotIn('ERROR', sql_msg, '执行失败' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.DROP_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name_1} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.table_name_1, sql_msg.splitlines()[-3],
                      '执行失败' + text)
        self.assertIn(self.user_name_1, sql_msg.splitlines()[-3],
                      '执行失败' + text)
        self.assertIn(self.view_name_1, sql_msg.splitlines()[-2],
                      '执行失败' + text)
        self.assertIn(self.user_name_1, sql_msg.splitlines()[-2],
                      '执行失败' + text)

        text = "-----step3:用户2创建依赖于表1的视图2;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop view if exists {self.view_name_2};" \
            f"create view {self.view_name_2} as " \
            f"select * from {self.table_name_1};"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name_2} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertNotIn('ERROR', sql_msg, '执行失败' + text)
        self.assertIn(self.constant.DROP_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name_2} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.view_name_2, sql_msg.splitlines()[-2], '执行失败' + text)
        self.assertIn(self.user_name_2, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step4:使用CASCADE级联删除所有依赖于被删除对象的对象;" \
               "expect:用户1所拥有的对象删除成功，且递归删除由用户2所拥有的对象-----"
        self.logger.info(text)
        sql_cmd = f"drop owned by {self.user_name_1} cascade;"
        sql_msg = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_OWNED_SUCCESS,
                      sql_msg, '执行失败' + text)
        sql_msg_1 = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name_1} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg_1)
        self.assertIn('No relations found', sql_msg_1, '执行失败' + text)
        sql_msg_2 = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name_2} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg_2)
        self.assertIn('No relations found', sql_msg_2, '执行失败' + text)

    def tearDown(self):
        text = "-----step5:删除用户，expect:删除成功-----"
        self.logger.info(text)
        drop_user = f"drop user {self.user_name_1} cascade;" \
            f"drop user {self.user_name_2} cascade;"
        drop_msg = self.primary_sh.execut_db_sql(drop_user)
        self.logger.info(drop_msg)
        self.assertEquals(2, drop_msg.count(
            self.constant.DROP_ROLE_SUCCESS_MSG), '执行失败' + text)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
